package com.zhss.dfs.namenode.server;

import com.zhss.dfs.datanode.server.DataNode;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 管理集群里的dataNode
 */
public class DataNodeManager {
    /**
     * 内存中维护dataNode的map
     */
    private Map<String,DataNodeInfo> dataNodes = new ConcurrentHashMap<String, DataNodeInfo>();

    public DataNodeManager(){
        new DataNodeAliveMonitor().start();
    }

    /**
     * datanode进行注册
     * @param ip
     * @param hostname
     * @return
     */
    public Boolean register(String ip,String hostname){
        DataNodeInfo dataNodeInfo =  new DataNodeInfo(ip,hostname);
        dataNodes.put(ip+"-"+hostname,dataNodeInfo);
        return true;
    }

    /**
     * datanode进行心跳
     * @param ip
     * @param hostname
     * @return
     */
    public Boolean heartbeat(String ip,String hostname){
        DataNodeInfo dataNode = dataNodes.get(ip+"-"+hostname);
        if(dataNode != null){
            dataNode.setLatestHeartbeatTime(System.currentTimeMillis());
        }
        return true;
    }

    /**
     * dataNode是否存活的监控线程
     */
    class DataNodeAliveMonitor extends Thread{
        @Override
        public void run() {
            try{
                while (true){
                    List<String> toRemoveDataNodes = new ArrayList<String>();
                    Iterator<DataNodeInfo> dataNodesIterator = dataNodes.values().iterator();
                    DataNodeInfo dataNode = null;
                    while(dataNodesIterator.hasNext()){
                        dataNode = dataNodesIterator.next();
                        if(System.currentTimeMillis() - dataNode.getLatestHeartbeatTime() > 90 * 1000){
                            toRemoveDataNodes.add(dataNode.getIp()+"-"+dataNode.getHostname());
                        }
                    }

                    if(!toRemoveDataNodes.isEmpty()){
                        for (String toRemoveDataNode : toRemoveDataNodes) {
                            dataNodes.remove(toRemoveDataNode);
                        }
                    }
                    Thread.sleep(30 * 1000);
                }
            }catch (Exception e){

            }
        }
    }
}
